package delay_queue

import (
	"channel/cond"
	"channel/queue"
	"context"
	"sort"
	"sync"
	"time"
)

type DelayQueue[T Delayable] struct {
	Queue    []T
	size     int
	notFull  *cond.Cond
	notEmpty *cond.Cond
	mu       *sync.Mutex
	resetCh  chan struct{}
}

func (p *DelayQueue[T]) isFull() bool {
	return p.size == len(p.Queue)
}

func (p *DelayQueue[T]) isEmpty() bool {
	return len(p.Queue) == 0
}

func (p *DelayQueue[T]) Enqueue(ctx context.Context, t T) error {
	if ctx.Err() != nil {
		return ctx.Err()
	}

	p.mu.Lock()
	if ctx.Err() != nil {
		return ctx.Err()
	}
	defer p.mu.Unlock()
	for p.isFull() {
		err := p.notFull.WaitV2(ctx)
		if err != nil {
			return err
		}
	}

	if len(p.Queue) > 0 && t.Delay().Before(p.Queue[0].Delay()) {
		select {
		case p.resetCh <- struct{}{}:
		default: // 如果 resetCh 已满，丢弃旧值
			<-p.resetCh
			p.resetCh <- struct{}{}
		}
	}

	p.Queue = append(p.Queue, t)
	sort.Slice(p.Queue, func(i, j int) bool {
		return p.Queue[i].Delay().Before(p.Queue[j].Delay())
	})
	if p.isFull() {
		p.notEmpty.Signal()
	}
	return nil
}

func (p *DelayQueue[T]) Dequeue(ctx context.Context) (T, error) {
	var t T
	for {
		if ctx.Err() != nil {
			return t, ctx.Err()
		}
		p.mu.Lock()
		if ctx.Err() != nil {
			return t, ctx.Err()
		}

		for p.isEmpty() {
			err := p.notEmpty.WaitV2(ctx)
			if err != nil {
				return t, err
			}
		}
		t = p.Queue[0]
		now := time.Now()
		if !t.Delay().After(now) {
			// 元素到期返回
			p.Queue = p.Queue[1:]
			if p.isEmpty() {
				p.notFull.Broadcast()
			}
			p.mu.Unlock()
			return t, nil
		}

		// 阻塞等待
		timer := time.NewTimer(t.Delay().Sub(now))
		p.mu.Unlock()

		select {
		case <-ctx.Done():

			timer.Stop()
			return t, ctx.Err()
		case <-timer.C:
			// 因为释放了锁，可能元素已经被人取走了，所以需要重新循环判断一下；如果合法，下次将会从!t.Delay().After(now)出去
			timer.Stop()
		case <-p.resetCh:
			// 来了新的超时时间更短的元素
			timer.Stop()
		}
	}
}
func NewDelayQueue[T Delayable](size int) *DelayQueue[T] {
	mu := &sync.Mutex{}
	return &DelayQueue[T]{
		size:     size,
		Queue:    make([]T, 0, size),
		notFull:  cond.NewCond(mu),
		notEmpty: cond.NewCond(mu),
		mu:       mu,
		resetCh:  make(chan struct{}, 1),
	}
}

type Delayable interface {
	Delay() time.Time
}

type DelayTest struct {
	time time.Time
	data any
}

func (d DelayTest) Delay() time.Time {
	return d.time
}

var _ queue.Queue[DelayTest] = (*DelayQueue[DelayTest])(nil)
